package cn.codeartist.springboot.component.rocketmq.proxy.impl;

import cn.codeartist.springboot.component.rocketmq.callback.SuccessSendCallback;
import cn.codeartist.springboot.component.rocketmq.proxy.MessageEntity;
import cn.codeartist.springboot.component.rocketmq.proxy.RocketMQProducer;
import cn.codeartist.springboot.component.rocketmq.proxy.RocketMQProducerEntity;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Optional;

/**
 * RocketMQ生产者发送消息抽象类封装
 *
 * @author 艾江南
 * @date 2020/8/13
 */
public abstract class AbstractRocketMQProducer extends RocketMQProducerEntity implements RocketMQProducer, BeanNameAware {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 创建Bean的名称
     */
    protected String beanName;
    /**
     * 多例的DefaultMQProducer实例
     */
    @Autowired
    protected DefaultMQProducer defaultMQProducer;

    /**
     * 初始化生产者Bean，使用多例创建不同的代理
     */
    @PostConstruct
    private void init() {
        try {
            defaultMQProducer = Optional.ofNullable(defaultMQProducer).orElse(new DefaultMQProducer());
            Optional.ofNullable(getNameSrvAddr()).ifPresent(defaultMQProducer::setNamesrvAddr);
            Optional.ofNullable(getGroup()).ifPresent(defaultMQProducer::setProducerGroup);
            Optional.ofNullable(getInstanceName()).ifPresent(defaultMQProducer::setInstanceName);
            defaultMQProducer.setRetryTimesWhenSendFailed(getRetryTimesWhenSendFailed());
            defaultMQProducer.start();
            logger.info("RocketMQ producer start with {}", defaultMQProducer);
        } catch (MQClientException e) {
            logger.error("RocketMQ producer register error.", e);
        }
    }

    @PreDestroy
    private void destroy() {
        if (defaultMQProducer != null) {
            defaultMQProducer.shutdown();
            logger.info("RocketMQ producer destroy with {}", defaultMQProducer);
        }
    }

    /**
     * 封装消息实体转换
     *
     * @param msg 消息封装
     * @return 消息
     */
    protected Message toMessage(MessageEntity msg) {
        Message message = new Message();
        message.setTopic(msg.getTopic());
        message.setTags(msg.getTag());
        message.setKeys(msg.getKey());
        message.setBody(JSON.toJSONString(msg.getData()).getBytes(StandardCharsets.UTF_8));
        Optional.ofNullable(msg.getGroup()).ifPresent(defaultMQProducer::setProducerGroup);
        logger.info("RocketMQ producer send message. bean:{}, group:{}, topic:{}, tag:{} message:{}",
                beanName, defaultMQProducer.getProducerGroup(), msg.getTopic(), msg.getTag(), JSON.toJSONString(msg));
        return message;
    }

    /**
     * 批量封装消息实体转换
     *
     * @param msgs 消息封装
     * @return 消息
     */
    protected List<Message> toMessage(Collection<MessageEntity> msgs) {
        List<Message> list = new ArrayList<>();
        for (MessageEntity msg : msgs) {
            Message message = toMessage(msg);
            list.add(message);
        }
        return list;
    }

    /**
     * 异步发送消息封装回调接口转换
     *
     * @param msg      消息体封装
     * @param callback 回调接口封装
     * @return 回调接口
     */
    protected SendCallback toSendCallback(MessageEntity msg, SuccessSendCallback callback) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                callback.onSuccess(sendResult);
                logger.info("RocketMQ producer send message async callback successful. bean:{}, group:{}, topic:{}, tag:{}, data:{}",
                        beanName, defaultMQProducer.getProducerGroup(), msg.getTopic(), msg.getTag(), JSON.toJSONString(msg.getData()));
            }

            @Override
            public void onException(Throwable e) {
                logger.error("RocketMQ producer send message async callback error. bean:{}, group:{}, topic:{}, tag:{}",
                        beanName, defaultMQProducer.getProducerGroup(), msg.getTopic(), msg.getTag(), e);
            }
        };
    }

    @Override
    public void setBeanName(String name) {
        this.beanName = name;
    }
}
